package conlon.rocketmq.mq.consumer;


import conlon.rocketmq.mq.listen.MessageListen;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

/**
 * @Author conlon
 * @Description 默认消费者
 * @Date 2021/1/6 20:06
 **/
@Component
@Slf4j
public class DefaultConsumer {

    private static final String DEMO_TAG = "tags01";



    @Value("${rocketmq.namesrvAddr:}")
    private String nameSrvAddr;

    @Value("${rocketmq.default.groupName:}")
    private String defaultGroupName;

    @Value("${rocketmq.default.topic01:}")
    private String topic01;

    /**
     * 通过构造函数 实例化对象
     * 通过测试发现  这里就算是不同组发过来的 消息 但是
     */
    @Bean
    public DefaultMQPushConsumer initDefaultConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(defaultGroupName);
        consumer.setNamesrvAddr(nameSrvAddr);
        //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //在监听类中增加两个消息处理类，当然可以增加更多，也就是往MessageListen类中的map集合放入处理类。
        MessageListen messageListener = new MessageListen();
        consumer.registerMessageListener(messageListener);
        //订阅主题和 标签（ * 代表所有标签)下信息
        consumer.subscribe(topic01 , "*");
        consumer.start();
        log.info(" DefaultConsumer 消费者 启动成功=======");
        return consumer;
    }
}